package org.btik.video.server.push.web;


import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.btik.app.tool.AsyncTaskExecutor;
import org.btik.context.thread.SystemControlAble;
import org.btik.video.server.VideoServer;
import org.btik.video.server.iface.DevChannel;
import org.btik.video.server.iface.HttpConstant;
import org.btik.video.server.iface.VideoChannel;
import org.btik.video.server.push.VideoChannelFactory;


import java.io.IOException;
import java.io.InputStream;
import java.net.*;
import java.util.Arrays;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;

/***
 * 以http mjpeg 合成视频流对VideoServer的实现
 * */
public class BioHttpVideoServer extends Thread implements HttpConstant, VideoServer, SystemControlAble {
    private final Log log = LogFactory.getLog(BioHttpVideoServer.class);

    private boolean runFlag = true;
    private AsyncTaskExecutor asyncTaskExecutor;

    private DevChannel devChannel;

    private int httpPort;

    private VideoChannelFactory videoChannelFactory;


    public void setAsyncTaskExecutor(AsyncTaskExecutor asyncTaskExecutor) {
        this.asyncTaskExecutor = asyncTaskExecutor;
    }

    public void setVideoChannelFactory(VideoChannelFactory videoChannelFactory) {
        this.videoChannelFactory = videoChannelFactory;
    }

    public void setHttpPort(int httpPort) {
        this.httpPort = httpPort;
    }

    public void setDevChannel(DevChannel devChannel) {
        this.devChannel = devChannel;
    }

    @Override
    public synchronized void start() {
        super.start();
        log.info("bio video server started");
    }

    @Override
    public void shutdown(String msg) {
        log.info("exit on:" + msg);
        runFlag = false;
    }

    @Override
    public void run() {
        try (ServerSocket serverSocket = new ServerSocket(httpPort)) {
            byte[] uri = new byte[URI_LEN];
            //channel 是 /{sn} 的形式 目前为 12位字符
            int channelIdLen = devChannel.channelIdLen();
            byte[] channel = new byte[channelIdLen + 1];
            while (runFlag) {
                Socket client = serverSocket.accept();
                InputStream inputStream = client.getInputStream();
                client.setSoTimeout(3000);

                try {
                    if (inputStream.read(uri) < URI_LEN) {
                        asyncTaskExecutor.execute(() -> do404(client));
                        continue;
                    }
                    // 判断uri
                    if (!Arrays.equals(uri, HttpConstant.uri)) {
                        asyncTaskExecutor.execute(() -> do404(client));
                        continue;
                    }
                    if (inputStream.read(channel) < channel.length) {
                        asyncTaskExecutor.execute(() -> do404(client));
                        continue;
                    }
                    String channelStr = new String(channel, 1,channelIdLen );
                    log.info("pre open" + new Date());
                    asyncTaskExecutor.execute(() -> doStreamOpen(client, channelStr));
                } catch (IOException e) {
                    disConnect(client, e);
                }

            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    private void doStreamOpen(Socket client, String channel) {
        try {
            MJPEGVideoChannel videoChannel = videoChannelFactory.getMJPEGVideoChannel(channel);
            if (null == videoChannel) {
                log.info("channel not exists:" + channel);
                do404(client);
                return;
            }
            client.setKeepAlive(true);
            client.setSoTimeout(300);
            videoChannel.joinChannel(client);
        } catch (IOException e) {
            log.error("join channel failed", e);
        }

    }

    private void do404(Socket client) {
        try {
            client.getOutputStream().write(NOT_FOUND);
        } catch (IOException e) {
            log.error(e.getMessage());
        } finally {
            disConnect(client, new Exception("404"));
        }

    }

    void disConnect(Socket socket, Exception e) {
        asyncTaskExecutor.execute(() -> {
            log.error(e.getMessage());
            try {
                log.error("close:" + socket.getRemoteSocketAddress());
                socket.close();
            } catch (IOException e0) {
                log.error(e.getMessage());
            }
        });

    }

}
